package timeandwindow;

import com.atguigu.pojo.OrderDetailEvent;
import com.atguigu.pojo.OrderEvent;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/*
*
*   IntervalJoin: 来自于两条流的相同key(join条件)的数据，以一条流中数据的时间为基准，
*   设置一个上界和下界， 形成一个时间范围，
    另外一条流对应的数据只要能落到该时间范围内，就可以Join成功
*   注意：运行代码时先监听启动端口，不然容易报错
*
* */

public class Flink15_IntervalJoin {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);

             //todo 模拟两条流:流1：订单流
         SingleOutputStreamOperator<OrderEvent> order1 = env.socketTextStream("hadoop102", 9999)
                 .map(
                         line -> {
                             String[] fields = line.split(",");
                             return new OrderEvent(fields[0].trim(), Long.valueOf(fields[1].trim()));
                         }
                 ).assignTimestampsAndWatermarks(
                         WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );
         order1.print("order1");

           //todo 模拟两条流:流2：订单详情流
         SingleOutputStreamOperator<OrderDetailEvent> order2 = env.socketTextStream("hadoop102", 8888)
                 .map(
                         line -> {
                             String[] fields = line.split(",");
                             return new OrderDetailEvent(fields[0].trim(),fields[1].trim(), Long.valueOf(fields[2].trim()));
                         }
                 ).assignTimestampsAndWatermarks(
                         WatermarkStrategy.<OrderDetailEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );
         order2.print("order2");

         order1.keyBy( OrderEvent::getId)
                 .intervalJoin(
                         order2.keyBy(OrderDetailEvent::getOrderId)
                 )
                 .between( Time.seconds(-2),Time.seconds(1))
                 .lowerBoundExclusive()//默认是不排除，排除下边界值
                 .upperBoundExclusive()//默认是不排除，排除上边界值
                 .process(
                         new ProcessJoinFunction<OrderEvent, OrderDetailEvent, String>() {
                             @Override
                             public void processElement(OrderEvent O1, OrderDetailEvent O2, ProcessJoinFunction<OrderEvent, OrderDetailEvent, String>.Context context, Collector<String> out) throws Exception {
                                 out.collect(O1 + "==" + O2);
                             }
                         }
                 )
                 .print("JOIN");



         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}
